/**
 * Tungsten Scale-Out Stack
 * Copyright (C) 2009-2010 Continuent Inc.
 * Contact: tungsten@continuent.org
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of version 2 of the GNU General Public License as
 * published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA
 *
 * Initial developer(s): Seppo Jaakola
 * Contributor(s): Stephane Giron
 */

package com.sync.extractor.mysql;

import java.io.IOException;
import java.sql.Timestamp;

import org.apache.log4j.Logger;

import com.sync.extractor.mysql.conversion.LittleEndianConversion;


/**
 * @author <a href="mailto:seppo.jaakola@continuent.com">Seppo Jaakola</a>
 * @author <a href="mailto:stephane.giron@continuent.com">Stephane Giron</a>
 * @version 1.0
 */

public abstract class LogEvent
{
    static Logger       logger              = Logger.getLogger(LogEvent.class);

    protected long      execTime;
    protected int       type;
    protected Timestamp when;
    protected int       serverId;

    protected int       logPos;
    protected int       flags;

    protected boolean   threadSpecificEvent = false;

    public LogEvent()
    {
        type = MysqlBinlog.START_EVENT_V3;
    }

    public LogEvent(byte[] buffer, FormatDescriptionLogEvent descriptionEvent,
            int eventType) throws MySQLExtractException
    {
        type = eventType;

        try
        {
            when = new Timestamp(1000 * LittleEndianConversion
                    .convert4BytesToLong(buffer, 0));
            serverId = (int) LittleEndianConversion.convert4BytesToLong(buffer,
                    MysqlBinlog.SERVER_ID_OFFSET);
            if (descriptionEvent.binlogVersion == 1)
            {
                logPos = 0;
                flags = 0;
                return;
            }

            /* 4.0 or newer */
            logPos = (int) LittleEndianConversion.convert4BytesToLong(buffer,
                    MysqlBinlog.LOG_POS_OFFSET);
            /*
             * If the log is 4.0 (so here it can only be a 4.0 relay log read by
             * the SQL thread or a 4.0 master binlog read by the I/O thread),
             * log_pos is the beginning of the event: we transform it into the
             * end of the event, which is more useful. But how do you know that
             * the log is 4.0: you know it if description_event is version 3and
             * you are not reading a Format_desc (remember that mysqlbinlog
             * starts by assuming that 5.0 logs are in 4.0 format, until it
             * finds a Format_desc).
             */

            if ((descriptionEvent.binlogVersion == 3)
                    && (buffer[MysqlBinlog.EVENT_TYPE_OFFSET] < MysqlBinlog.FORMAT_DESCRIPTION_EVENT)
                    && (logPos > 0))
            {
                /*
                 * If log_pos=0, don't change it. log_pos==0 is a marker to mean
                 * "don't change rli->group_master_log_pos" (see
                 * inc_group_relay_log_pos()). As it is unreal log_pos, adding
                 * the event len's is nonsense. For example, a fake Rotate event
                 * should not have its log_pos (which is 0) changed or it will
                 * modify Exec_master_log_pos in SHOW SLAVE STATUS, displaying a
                 * nonsense value of (a non-zero offset which does not exist in
                 * the master's binlog, so which will cause problems if the user
                 * uses this value in CHANGE MASTER).
                 */
                logPos += LittleEndianConversion.convert4BytesToLong(buffer,
                        MysqlBinlog.EVENT_LEN_OFFSET);
            }
            if (logger.isDebugEnabled())
                logger.debug("log_pos: " + logPos);

            flags = LittleEndianConversion.convert2BytesToInt(buffer,
                    MysqlBinlog.FLAGS_OFFSET);
            /*
             * TODO LOG_EVENT_THREAD_SPECIFIC_F = 0x4 (New in 4.1.0) Used only
             * by mysqlbinlog (not by the replication code at all) to be able to
             * deal properly with temporary tables. mysqlbinlog displays events
             * from the binary log in printable format, so that you can feed the
             * output into mysql (the command-line interpreter), to achieve
             * incremental backup recovery.
             */
            threadSpecificEvent = ((flags & MysqlBinlog.LOG_EVENT_THREAD_SPECIFIC_F) == MysqlBinlog.LOG_EVENT_THREAD_SPECIFIC_F);
            if (logger.isDebugEnabled())
                logger.debug("Event is thread-specific = "
                        + threadSpecificEvent);

            if ((buffer[MysqlBinlog.EVENT_TYPE_OFFSET] == MysqlBinlog.FORMAT_DESCRIPTION_EVENT)
                    || (buffer[MysqlBinlog.EVENT_TYPE_OFFSET] == MysqlBinlog.ROTATE_EVENT))
            {
                /*
                 * These events always have a header which stops here (i.e.
                 * their header is FROZEN).
                 */
                /*
                 * Initialization to zero of all other Log_event members as
                 * they're not specified. Currently there are no such members;
                 * in the future there will be an event UID (but
                 * Format_description and Rotate don't need this UID, as they
                 * are not propagated through --log-slave-updates (remember the
                 * UID is used to not play a query twice when you have two
                 * masters which are slaves of a 3rd master). Then we are done.
                 */
                return;
            }
            /*
             * otherwise, go on with reading the header from buffer (nothing for
             * now)
             */
        }
        catch (IOException e)
        {
            logger.error("Cannot create log event: " + e);
            throw new MySQLExtractException("log event create failed");
        }
    }

    public long getExecTime()
    {
        return execTime;
    }

    public Timestamp getWhen()
    {
        return when;
    }

    public static LogEvent readLogEvent(boolean parseStatements,
            byte[] buffer, int eventLength,
            FormatDescriptionLogEvent descriptionEvent,
            boolean useBytesForString) throws MySQLExtractException
    {
        LogEvent event = null;
        
        //System.out.println(buffer[MysqlBinlog.EVENT_TYPE_OFFSET]);

        switch (buffer[MysqlBinlog.EVENT_TYPE_OFFSET ]) 
        {
            case MysqlBinlog.QUERY_EVENT :
                event = new QueryLogEvent(buffer, eventLength,
                        descriptionEvent, parseStatements, useBytesForString);
                break;
            case MysqlBinlog.LOAD_EVENT :
                logger.warn("Skipping unsupported LOAD_EVENT");
                // ev = new Load_log_event(buf, event_len, description_event);
                break;
            case MysqlBinlog.NEW_LOAD_EVENT :
                logger.warn("Skipping unsupported NEW_LOAD_EVENT");
                // ev = new Load_log_event(buf, event_len, description_event);
                break;
            case MysqlBinlog.ROTATE_EVENT :
                event = new RotateLogEvent(buffer, eventLength,
                        descriptionEvent);
                break;
            case MysqlBinlog.SLAVE_EVENT : /* can never happen (unused event) */
                logger.warn("Skipping unsupported SLAVE_EVENT");
                // ev = new Slave_log_event(buf, event_len);
                break;
            case MysqlBinlog.CREATE_FILE_EVENT :
                logger.warn("Skipping unsupported CREATE_FILE_EVENT");
                // ev = new Create_file_log_event(buf, event_len,
                // description_event);
                break;
            case MysqlBinlog.APPEND_BLOCK_EVENT :
                logger.debug("reading APPEND_BLOCK_EVENT");
                event = new AppendBlockLogEvent(buffer, eventLength,
                        descriptionEvent);
                // ev = new Append_block_log_event(buf, event_len,
                // description_event);
                break;
            case MysqlBinlog.DELETE_FILE_EVENT :
                logger.debug("reading DELETE_FILE_EVENT");
                event = new DeleteFileLogEvent(buffer, eventLength,
                        descriptionEvent);
                // ev = new Delete_file_log_event(buf, event_len,
                // description_event);
                break;
            case MysqlBinlog.EXEC_LOAD_EVENT :
                logger.warn("Skipping unsupported EXEC_LOAD_EVENT");
                // ev = new Execute_load_log_event(buf, event_len,
                // description_event);
                break;
            case MysqlBinlog.START_EVENT_V3 :
                /* this is sent only by MySQL <=4.x */
                logger.warn("Skipping unsupported START_EVENT_V3");
                // ev = new Start_log_event_v3(buf, description_event);
                break;
            case MysqlBinlog.STOP_EVENT :
                event = new StopLogEvent(buffer, eventLength, descriptionEvent);
                break;
            case MysqlBinlog.INTVAR_EVENT :
                logger.debug("extracting INTVAR_EVENT");
                event = new IntvarLogEvent(buffer, eventLength,
                        descriptionEvent);
                break;
            case MysqlBinlog.XID_EVENT :
                event = new XidLogEvent(buffer, eventLength, descriptionEvent);
                break;
            case MysqlBinlog.RAND_EVENT :
                event = new RandLogEvent(buffer, eventLength, descriptionEvent);
                break;
            case MysqlBinlog.USER_VAR_EVENT :
                event = new UserVarLogEvent(buffer, eventLength,
                        descriptionEvent);
                break;
            case MysqlBinlog.FORMAT_DESCRIPTION_EVENT :
                event = new FormatDescriptionLogEvent(buffer, eventLength,
                        descriptionEvent);
                break;
            case MysqlBinlog.PRE_GA_WRITE_ROWS_EVENT :
                logger.warn("Skipping unsupported PRE_GA_WRITE_ROWS_EVENT");
                // ev = new Write_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            case MysqlBinlog.PRE_GA_UPDATE_ROWS_EVENT :
                logger.warn("Skipping unsupported PRE_GA_UPDATE_ROWS_EVENT");
                // ev = new Update_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            case MysqlBinlog.PRE_GA_DELETE_ROWS_EVENT :
                logger.warn("Skipping unsupported PRE_GA_DELETE_ROWS_EVENT");
                // ev = new Delete_rows_log_event_old(buf, event_len,
                // description_event);
                break;
            case MysqlBinlog.WRITE_ROWS_EVENT :
                logger.debug("reading WRITE_ROWS_EVENT");
                event = new WriteRowsLogEvent(buffer, eventLength,
                        descriptionEvent, useBytesForString);
                break;
            case MysqlBinlog.UPDATE_ROWS_EVENT :
                logger.debug("reading UPDATE_ROWS_EVENT");
                event = new UpdateRowsLogEvent(buffer, eventLength,
                        descriptionEvent, useBytesForString);
                break;
            case MysqlBinlog.DELETE_ROWS_EVENT :
                logger.debug("reading DELETE_ROWS_EVENT");
                event = new DeleteRowsLogEvent(buffer, eventLength,
                        descriptionEvent, useBytesForString);
                break;
            case MysqlBinlog.TABLE_MAP_EVENT :
                logger.debug("reading TABLE_MAP_EVENT");
                event = new TableMapLogEvent(buffer, eventLength,
                        descriptionEvent);
                break;
            case MysqlBinlog.BEGIN_LOAD_QUERY_EVENT :
                logger.debug("reading BEGIN_LOAD_QUERY_EVENT");
                event = new BeginLoadQueryLogEvent(buffer, eventLength,
                        descriptionEvent);
                break;
            case MysqlBinlog.EXECUTE_LOAD_QUERY_EVENT :
                logger.debug("reading EXECUTE_LOAD_QUERY_EVENT");
                event = new ExecuteLoadQueryLogEvent(buffer, eventLength,
                        descriptionEvent, parseStatements);
                break;
            case MysqlBinlog.INCIDENT_EVENT :
                logger.warn("Skipping unsupported INCIDENT_EVENT");
                // ev = new Incident_log_event(buf, event_len,
                // description_event);
                break;
            default :
                logger.warn("Skipping unrecognized binlog event type "
                        + (buffer[MysqlBinlog.EVENT_TYPE_OFFSET]));
        }
        return event;
    }

    protected String hexdump(byte[] buffer, int offset)
    {
        StringBuffer dump = new StringBuffer();
        if ((buffer.length - offset) > 0)
        {
            dump.append(String.format("%02x", buffer[offset]));
            for (int i = offset + 1; i < buffer.length; i++)
            {
                dump.append("_");
                dump.append(String.format("%02x", buffer[i]));
            }
        }
        return dump.toString();
    }

    protected String hexdump(byte[] buffer, int offset, int length)
    {
        StringBuffer dump = new StringBuffer();

        if (buffer.length >= offset + length)
        {
            dump.append(String.format("%02x", buffer[offset]));
            for (int i = offset + 1; i < offset + length; i++)
            {
                dump.append("_");
                dump.append(String.format("%02x", buffer[i]));
            }
        }
        return dump.toString();
    }

}
